package com.example.everying.service.content.impl;

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.example.everying.mapper.content.ActivityArticleMapper;
import com.example.everying.mapper.content.ActivityThumbMapper;
import com.example.everying.mapper.user.UserMapper;
import com.example.everying.model.entity.content.ActivityArticle;
import com.example.everying.model.entity.content.ActivityThumb;
// import com.example.everying.model.entity.user.User;
import com.example.everying.model.entity.user.UserEntity;
import com.example.everying.model.vo.content.ActivityArticleVo;
import com.example.everying.service.content.ActivityThumbService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.core.RedisOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.SessionCallback;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * @program: everything
 * @description: 文章点赞实现层
 * @author: zw
 * @create: 2025-05-22 15:27
 **/
@Service
@Slf4j
public class ActivityThumbServiceImpl extends ServiceImpl<ActivityThumbMapper, ActivityThumb> implements ActivityThumbService {

    private static final String ARTICLE_LIKES_KEY = "article:likes:%s";
    private static final String ARTICLE_LIKES_COUNT_KEY = "article:likes:count:%s";
    private static final String PENDING_SYNC_KEY = "article:likes:pending";
    
    // 消息通知相关常量
    private static final String USER_MESSAGES_KEY = "user:messages:%s";
    private static final String MESSAGE_DETAIL_KEY = "message:detail:%s";
    private static final String USER_UNREAD_COUNT_KEY = "user:unread:count:%s";
    private static final String MESSAGE_ID_COUNTER = "message:id:counter";

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    @Autowired
    private ActivityArticleMapper articleMapper;
    
    @Autowired
    private UserMapper userMapper;

    @Override
    public boolean likeArticle(String articleId, String userId) {
        String articleLikesKey = String.format(ARTICLE_LIKES_KEY, articleId);
        String likesCountKey = String.format(ARTICLE_LIKES_COUNT_KEY, articleId);

        // 检查是否已经点赞过
        if (hasLiked(articleId, userId)) {
            return false; // 已经点赞过，不重复点赞
        }

        // 使用 Redis 事务保证操作原子性
        redisTemplate.execute(new SessionCallback<Object>() {
            @Override
            public <K, V> Object execute(RedisOperations<K, V> operations) throws DataAccessException {
                operations.multi();
                // 添加用户到点赞集合
                operations.opsForSet().add((K) articleLikesKey, (V) userId);
                // 点赞数加1
                operations.opsForValue().increment((K) likesCountKey, 1);
                // 将文章ID添加到待同步集合
                operations.opsForSet().add((K) PENDING_SYNC_KEY, (V) articleId);
                return operations.exec();
            }
        });

        // 异步创建点赞消息通知
        createLikeMessage(articleId, userId);

        return true;
    }

    @Override
    public boolean unlikeArticle(String articleId, String userId) {
        String articleLikesKey = String.format(ARTICLE_LIKES_KEY, articleId);
        String likesCountKey = String.format(ARTICLE_LIKES_COUNT_KEY, articleId);

        // 使用 Redis 事务保证操作原子性
        redisTemplate.execute(new SessionCallback<Object>() {
            @Override
            public <K, V> Object execute(RedisOperations<K, V> operations) throws DataAccessException {
                operations.multi();
                // 从点赞集合中移除用户
                operations.opsForSet().remove((K) articleLikesKey, (V) userId);
                // 点赞数减1
                operations.opsForValue().decrement((K) likesCountKey, 1);
                // 将文章ID添加到待同步集合
                operations.opsForSet().add((K) PENDING_SYNC_KEY, (V) articleId);
                return operations.exec();
            }
        });
        return true;
    }

    @Override
    public boolean hasLiked(String articleId, String userId) {
        String articleLikesKey = String.format(ARTICLE_LIKES_KEY, articleId);
        return redisTemplate.opsForSet().isMember(articleLikesKey, userId);
    }

    @Override
    public long getLikesCount(String articleId) {
        String likesCountKey = String.format(ARTICLE_LIKES_COUNT_KEY, articleId);
        Object countObj = redisTemplate.opsForValue().get(likesCountKey);
        return countObj != null ? Long.parseLong(countObj.toString()) : 0;
    }

    @Override
    @Async("asyncExecutor")
    @Transactional
    public void syncLikesToDatabase() {
        String pendingSyncKey = PENDING_SYNC_KEY;

        // 获取所有待同步的文章ID
        Set<Object> articleIds = redisTemplate.opsForSet().members(pendingSyncKey);

        if (articleIds == null || articleIds.isEmpty()) {
            return;
        }

        for (Object articleIdObj : articleIds) {
            String articleId = articleIdObj.toString();
            syncArticleLikes(articleId);
        }
    }

    private void syncArticleLikes(String articleId) {
        String articleLikesKey = String.format(ARTICLE_LIKES_KEY, articleId);
        String likesCountKey = String.format(ARTICLE_LIKES_COUNT_KEY, articleId);
        try {
            // 1. 获取Redis中的点赞用户
            Set<Object> redisUserIds = redisTemplate.opsForSet().members(articleLikesKey);
            Set<String> redisUserSet = redisUserIds.stream()
                    .map(Object::toString)
                    .collect(Collectors.toSet());

            // 2. 获取数据库中已有的点赞用户
            List<ActivityThumb> dbLikes = baseMapper.selectList(
                    new QueryWrapper<ActivityThumb>().eq("article_id", articleId)
            );
            Set<String> dbUserSet = dbLikes.stream()
                    .map(ActivityThumb::getUserId)
                    .collect(Collectors.toSet());

            // 3. 计算需要新增和删除的用户
            Set<String> addUsers = redisUserSet.stream()
                    .filter(u -> !dbUserSet.contains(u))
                    .collect(Collectors.toSet());
            Set<String> deleteUsers = dbUserSet.stream()
                    .filter(u -> !redisUserSet.contains(u))
                    .collect(Collectors.toSet());

            // 4. 增量更新：删除需要移除的点赞
            if (!deleteUsers.isEmpty()) {
                baseMapper.delete(new QueryWrapper<ActivityThumb>()
                        .eq("article_id", articleId)
                        .in("user_id", deleteUsers));
            }

            // 5. 增量更新：新增需要添加的点赞
            if (!addUsers.isEmpty()) {
                List<ActivityThumb> newLikes = addUsers.stream().map(userId -> {
                    ActivityThumb like = new ActivityThumb();
                    like.setArticleId(articleId);
                    like.setUserId(userId);
                    like.setCreateTime(LocalDateTime.now());
                    like.setUpdateTime(LocalDateTime.now());
                    return like;
                }).collect(Collectors.toList());
                baseMapper.insertBatch(newLikes);
            }

            // 6. 更新点赞数（复用现有逻辑）
            Integer count = redisUserSet.size();
            updateArticleLikesCount(articleId, count);

            // 7. 从待同步集合中移除
            redisTemplate.opsForSet().remove(PENDING_SYNC_KEY, articleId);

        } catch (Exception e) {
            log.error("同步文章点赞数据失败，文章ID: {}", articleId, e);
        }
    }

    private void updateArticleLikesCount(String articleId, Integer count) {
        log.info("准备更新文章点赞数，articleId: {}, count: {}", articleId, count);
        // 根据 id 查询现有记录
        LambdaUpdateWrapper<ActivityArticle> wrapper = new LambdaUpdateWrapper<>();
        wrapper.eq(ActivityArticle::getId, articleId)
                .set(ActivityArticle::getThumbs, count);

        articleMapper.update(null, wrapper);
        log.info("更新文章点赞数完成，articleId: {}, count: {}", articleId, count);
    }

    /**
     * 文章点赞列表查询
     * @param userId
     * @return
     */
    @Override
    public List<ActivityArticleVo> getUserLikes(String userId) {
        return baseMapper.selectUserId(userId);
    }

    @Override
    public Map<String, Boolean> batchHasLiked(List<String> articleIds, String userId) {
        Map<String, Boolean> result = new HashMap<>();
        
        if (articleIds == null || articleIds.isEmpty()) {
            return result;
        }
        
        // 使用管道操作提高Redis性能
        List<Object> results = redisTemplate.executePipelined(new SessionCallback<Object>() {
            @Override
            public <K, V> Object execute(RedisOperations<K, V> operations) throws DataAccessException {
                for (String articleId : articleIds) {
                    String articleLikesKey = String.format(ARTICLE_LIKES_KEY, articleId);
                    operations.opsForSet().isMember((K) articleLikesKey, (V) userId);
                }
                return null;
            }
        });
        
        // 组装结果
        for (int i = 0; i < articleIds.size(); i++) {
            result.put(articleIds.get(i), (Boolean) results.get(i));
        }
        
        return result;
    }

    @Override
    public Map<String, Long> batchGetLikesCount(List<String> articleIds) {
        Map<String, Long> result = new HashMap<>();
        
        if (articleIds == null || articleIds.isEmpty()) {
            return result;
        }
        
        // 使用管道操作提高Redis性能
        List<Object> results = redisTemplate.executePipelined(new SessionCallback<Object>() {
            @Override
            public <K, V> Object execute(RedisOperations<K, V> operations) throws DataAccessException {
                for (String articleId : articleIds) {
                    String likesCountKey = String.format(ARTICLE_LIKES_COUNT_KEY, articleId);
                    operations.opsForValue().get((K) likesCountKey);
                }
                return null;
            }
        });
        
        // 组装结果
        for (int i = 0; i < articleIds.size(); i++) {
            Object countObj = results.get(i);
            Long count = countObj != null ? Long.parseLong(countObj.toString()) : 0L;
            result.put(articleIds.get(i), count);
        }
        
        return result;
    }

    /**
     * 创建点赞消息通知
     */
    @Async("asyncExecutor")
    private void createLikeMessage(String articleId, String senderId) {
        try {
            // 获取文章信息，判断是否需要发送通知
            ActivityArticle article = articleMapper.selectById(articleId);
            if (article != null && !article.getUserId().equals(senderId)) {
                // 不给自己发送点赞通知
                createMessageInRedis(senderId, article.getUserId(), articleId, "like", null);
            }
        } catch (Exception e) {
            log.error("创建点赞消息失败", e);
        }
    }

    /**
     * 在Redis中创建消息
     */
    private void createMessageInRedis(String senderId, String receiverId, String articleId, 
                                    String messageType, String commentId) {
        try {
            // 生成消息ID
            Long messageId = redisTemplate.opsForValue().increment(MESSAGE_ID_COUNTER);
            String messageKey = String.format(MESSAGE_DETAIL_KEY, messageId);
            String userMessagesKey = String.format(USER_MESSAGES_KEY, receiverId);
            String unreadCountKey = String.format(USER_UNREAD_COUNT_KEY, receiverId);

            // 消息详情
            Map<String, Object> messageDetail = new HashMap<>();
            messageDetail.put("id", messageId.toString());
            messageDetail.put("senderId", senderId);
            messageDetail.put("receiverId", receiverId);
            messageDetail.put("articleId", articleId);
            messageDetail.put("messageType", messageType);
            messageDetail.put("commentId", commentId);
            messageDetail.put("isRead", false);
            messageDetail.put("createTime", System.currentTimeMillis());

            // 使用Redis事务保证原子性
            redisTemplate.execute(new SessionCallback<Object>() {
                @Override
                public <K, V> Object execute(RedisOperations<K, V> operations) throws DataAccessException {
                    operations.multi();
                    
                    // 保存消息详情
                    operations.opsForHash().putAll((K) messageKey, (Map<? extends K, ? extends V>) messageDetail);
                    operations.expire((K) messageKey, 30, TimeUnit.DAYS);
                    
                    // 添加到用户消息列表（按时间倒序）
                    operations.opsForZSet().add((K) userMessagesKey, (V) messageId.toString(), 
                                             -System.currentTimeMillis());
                    operations.expire((K) userMessagesKey, 30, TimeUnit.DAYS);
                    
                    // 增加未读消息数
                    operations.opsForValue().increment((K) unreadCountKey, 1);
                    operations.expire((K) unreadCountKey, 30, TimeUnit.DAYS);
                    
                    return operations.exec();
                }
            });
            
        } catch (Exception e) {
            log.error("创建Redis消息失败", e);
        }
    }

    /**
     * 获取用户消息列表（合并评论和点赞消息）
     */
    public List<Map<String, Object>> getMessageList(String userId, Integer page, Integer limit) {
        try {
            String userMessagesKey = String.format(USER_MESSAGES_KEY, userId);
            
            // 分页获取消息ID列表
            long start = (page - 1) * limit;
            long end = start + limit - 1;
            
            Set<Object> messageIds = redisTemplate.opsForZSet().range(userMessagesKey, start, end);
            
            if (messageIds == null || messageIds.isEmpty()) {
                return new ArrayList<>();
            }
            
            // 批量获取消息详情
            List<Map<String, Object>> messages = new ArrayList<>();
            for (Object messageId : messageIds) {
                String messageKey = String.format(MESSAGE_DETAIL_KEY, messageId);
                Map<Object, Object> messageDetail = redisTemplate.opsForHash().entries(messageKey);
                
                if (!messageDetail.isEmpty()) {
                    Map<String, Object> message = new HashMap<>();
                    messageDetail.forEach((k, v) -> message.put(k.toString(), v));
                    
                    // 获取发送者信息
                    String senderId = message.get("senderId").toString();
                    UserEntity sender = userMapper.selectById(senderId);
                    if (sender != null) {
                        Map<String, Object> senderInfo = new HashMap<>();
                        senderInfo.put("id", sender.getId());
                        senderInfo.put("userName", sender.getUserName());
                        senderInfo.put("avatarUrl", sender.getAvatarUrl());
                        message.put("senderInfo", senderInfo);
                    }
                    
                    // 获取文章信息
                    String articleId = message.get("articleId").toString();
                    ActivityArticle article = articleMapper.selectById(articleId);
                    if (article != null) {
                        Map<String, Object> articleInfo = new HashMap<>();
                        articleInfo.put("id", article.getId());
                        articleInfo.put("title", article.getTitle());
                        articleInfo.put("type", article.getType());
                        message.put("activityArticle", articleInfo);
                    }
                    
                    messages.add(message);
                }
            }
            
            return messages;
            
        } catch (Exception e) {
            log.error("获取消息列表失败", e);
            return new ArrayList<>();
        }
    }

    /**
     * 获取未读消息数
     */
    public Integer getUnreadCount(String userId) {
        try {
            String unreadCountKey = String.format(USER_UNREAD_COUNT_KEY, userId);
            Object count = redisTemplate.opsForValue().get(unreadCountKey);
            return count != null ? Integer.parseInt(count.toString()) : 0;
        } catch (Exception e) {
            log.error("获取未读消息数失败", e);
            return 0;
        }
    }

    /**
     * 标记消息为已读
     */
    public void markAsRead(String messageId, String userId) {
        try {
            String messageKey = String.format(MESSAGE_DETAIL_KEY, messageId);
            String unreadCountKey = String.format(USER_UNREAD_COUNT_KEY, userId);
            
            // 检查消息是否未读
            Object isRead = redisTemplate.opsForHash().get(messageKey, "isRead");
            if (isRead != null && !(Boolean) isRead) {
                // 标记为已读
                redisTemplate.opsForHash().put(messageKey, "isRead", true);
                // 减少未读数
                redisTemplate.opsForValue().decrement(unreadCountKey, 1);
            }
        } catch (Exception e) {
            log.error("标记消息已读失败", e);
        }
    }

    /**
     * 批量标记为已读
     */
    public void markAllAsRead(String userId) {
        try {
            String userMessagesKey = String.format(USER_MESSAGES_KEY, userId);
            String unreadCountKey = String.format(USER_UNREAD_COUNT_KEY, userId);
            
            // 获取所有消息ID
            Set<Object> messageIds = redisTemplate.opsForZSet().range(userMessagesKey, 0, -1);
            
            if (messageIds != null && !messageIds.isEmpty()) {
                for (Object messageId : messageIds) {
                    String messageKey = String.format(MESSAGE_DETAIL_KEY, messageId);
                    redisTemplate.opsForHash().put(messageKey, "isRead", true);
                }
                
                // 重置未读数为0
                redisTemplate.opsForValue().set(unreadCountKey, 0);
            }
        } catch (Exception e) {
            log.error("批量标记消息已读失败", e);
        }
    }
}